﻿#include <iostream>
#include <thread>

#include <zmq.hpp>

#include "../../inc/helper.hpp"

const static int g_worker_number = 5;
const static int g_client_number = 3;

void client_task()
{
	zmq::context_t ctx;
	zmq::socket_t client(ctx, zmq::socket_type::dealer);

	std::string id = gen_random_id();
	client.setsockopt(ZMQ_IDENTITY, id.data(), id.size());
	client.connect("tcp://localhost:5570");

	std::vector<zmq::pollitem_t> items = {
		{client, 0, ZMQ_POLLIN, 0}
	};

	Time next_request_time = Now();
	int request_number = 0;
	while (true)
	{
		zmq::poll(items, 10);
		if (items[0].revents & ZMQ_POLLIN)
		{
			auto msgs = zmsg_recv(client);
			info("{} : {}", id, to_string(msgs.back()));
		}

		if (Now() >= next_request_time)
		{
			send_string(client, fmt::format("request #{}", ++request_number));
			next_request_time = Now() + Seconds(1);
		}
	}
}

void server_worker(zmq::context_t& ctx)
{
	zmq::socket_t worker(ctx, zmq::socket_type::dealer);
	worker.connect("inproc://backend");

	std::random_device rd;
	std::uniform_int_distribution<int> reply_num(0, 5);
	std::uniform_int_distribution<int> sleep_len(0, 1000);
	while (true)
	{
		auto msgs = zmsg_recv(worker);

		int replies = reply_num(rd);
		for (int i = 0; i < replies; ++i)
		{
			int sleep_ms = sleep_len(rd);
			std::this_thread::sleep_for(std::chrono::milliseconds(sleep_ms));

			auto copied_msgs = copy(msgs);
			zmsg_send(worker, copied_msgs);
		}
	}
}

void server_task()
{
	zmq::context_t ctx;

	zmq::socket_t frontend(ctx, zmq::socket_type::router);
	frontend.bind("tcp://*:5570");

	zmq::socket_t backend(ctx, zmq::socket_type::dealer);
	backend.bind("inproc://backend");

	for (int i = 0; i < g_worker_number; ++i)
	{
		std::thread t(server_worker, std::ref(ctx));
		t.detach();
	}

	zmq::proxy(frontend, backend, NULL);
	return;
}

int main()
{
	for (int i = 0; i < g_client_number; ++i)
	{
		std::thread t(client_task);
		t.detach();
	}
	std::thread t(server_task);
	t.detach();

	pause();
}